package com.crr;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;

public class Productor {
    public static void main(String[] args) {
        //1创建连接工厂 创建连接
        ConnectionFactory factory = new ConnectionFactory();
        factory.setVirtualHost("/");
        factory.setPort(5672);
        factory.setUsername("user");
        factory.setPassword("chen1116.");
        // factory.setHost("127.0.0.1");

        //2创建连接
        try (Connection conn = factory.newConnection();
             //3创建信道
             Channel channel = conn.createChannel()) {
            //4 启用发布确认
            channel.confirmSelect();
            //存储消息
            ConcurrentNavigableMap msgMap = new ConcurrentSkipListMap();
            //接收服务器确认的消息
            channel.addConfirmListener((deliveryTag, multiple) -> {
                System.out.println("服务器确认消息：" + deliveryTag);
                if(multiple){
                    //为true 批量确认 小于等于当前数的都是服务器确认收到的
                    ConcurrentNavigableMap tmpMap = msgMap.headMap(deliveryTag);//返回一个子集
                    System.out.println("tmpMap"+tmpMap.size());

                }else {
                    //为false 则消息为单个确认
                    String tmp=msgMap.get(deliveryTag).toString();
                    System.out.println("消息发布成功"+tmp);
                    //移除消息
                    msgMap.remove(deliveryTag);
                }
            }, (deliveryTag, multiple) -> {
                System.out.println("消息发送失败：" + msgMap.get(deliveryTag));
            });
            //声明队列
            String queueName = "confirm_async";

            channel.queueDeclare(queueName, false, false, false, null);


            long begin = System.currentTimeMillis();
            int tag = 0;
            //发送消息
            for (int i = 1; i <= 999; i++) {
                //获取消息发布的序列号
                long seq = channel.getNextPublishSeqNo();
                //要发送的消息
                String msg="消息"+ i;
                //将序号和消息关联起来
                msgMap.put(seq,msg);
                channel.basicPublish("", queueName, null, ("发布确认" + i).getBytes());

            }
            long end = System.currentTimeMillis();

            System.out.println("耗费时间：" + (end - begin));
            System.in.read();
            System.out.println("消息已发送,信息已收到");
        } catch (IOException e) {
            throw new RuntimeException(e);
        } catch (TimeoutException e) {
            throw new RuntimeException(e);
        }

    }

}
